# 连接和传输
"""
几种传输amqp, redis, qpid, librabbitmq等。
"""
# 使用默认传输连接
from kombu import Connection, Producer

connection = Connection('amqp://guest@localhost:5672//')

# 网址：连接的参数格式可以使url
#transport://userid:password@hostname:port/virtual_host

"""
AMQP有四种可供AMQP使用：
1、pyamqp使用纯Python库amqp,与kombu一起自动安装。
2、librabbitmq使用c语言编写的高性能传输。这需要librrabbitqp
安装python包，他可以自动编译C库。
等
"""

"""创建生产者Connection"""
producer = connection.Producer()

"""您也可以直接实例化，他将通道或者连接作为参数"""
with Connection('amqp://') as conn:
    with conn.channel() as channel:
        producer = Producer(channel)

"""大对数情况下，将从连接池获取连接，并且此链接可能过时，
或者可能在发送消息时丢失.使用重试时处理这些间歇性故障的好方法：
"""
producer.publish({"hello": "hello"}, retry=True)

# 另外一个重试可以指定，这是由支持的参数的词典retry_over_time()功能：
producer.publish(
    {"hello": "hello",
     "retry": True,
     "retry_policy": {
         "interval_start": 0,
         "interval_step": 2,
         "interval_max": 30,
         "max_retries": 30,
     }

    }
)

"""
该declare参数允许在传递必须在发送消息之前声明实体列表，这在使用retry标志时
尤其重要，因为代理可能早重试期间实际重启，在这种情况下，非持久实体被移除。
假设正在编写任务队列， 并且没有启动worker，因此未声明队列，在这种情况下，
需要定义转换，并声明队列，以便在工作人员脱机时将消息传递到队列。
"""
from kombu import Exchange, Queue
task_queue = Queue('tasks', Exchange('tasks'), routing_key='tasks')
producer.publish(
    {"hello": "hello"},
    retry=True,
    exchange=task_queue.exchange,
    routing_key=task_queue.routing_key,
    declare=[task_queue]
)

# 使用匿名交换绕过路由
"""
您可以通过使用"anon-exchange"直接传递到队列，绕过代理路由机制：
将exchange参数设置为空字符串，并将路由键设置为队列的名称。
"""
producer.publish(
    {"hello": 'hello'},
    exhange='',
    routing_key=task_queue.name,
)

# 序列化
"""
将非字符串传递个发布时，json是默认的序列化程序，也可以指定其他的序列化程序
"""
producer.publish({"hello": "hello"}, serializer="pickle")

# 基础知识
"""
在Consumer需要一个连接(或通道)和队列从消耗的列表里面，几个消费者可以混合来
使用来自不同的渠道，因为他们都绑定到同一连接，并且drain_events会从该连接上的
所有通道中排出事件。
"""
#从单个消费者中排出事件：
from kombu import Consumer
with Consumer(connection, queues=task_queue, accept=["json"]):
    connection.drain_events(timeout=1)

#从几个消费者中排出事件
from kombu.utils.compat import nested
with connection.channel(), connection.channel() as (channel,channel2):
    with nested(Consumer(channel, queues=None, accept=["json"]),
                Consumer(channel2, queues=None, accept=["json"])):
        connection.drain_events(timeout=1)

# 或者使用ConsumerMixin
from kombu.mixins import ConsumerMixin

class C(ConsumerMixin):
    def __init__(self, connection):
        self.connection = connection

    def get_consumers(self, Consumer, channel):
        return[
            Consumer(queues=task_queue, callback=[self.on_message],
                     accept=["json"])
        ]

    def on_message(self, body, message):
        print('RECEIVED MESSAGE:{0!r}'.format(body))
        message.ack()


C(connection).run()

# 并在此使用多个频道
from kombu import Consumer
from kombu.mixins import ConsumerMixin

class C(ConsumerMixin):
    channel2 = None

    def __init__(self, connection):
        self.connection = connection

    def get_consumers(self, _, default_channel):
        self.channel2 = default_channel.connection.channel()
        return [Consumer(default_channel, queues1=task_queue,
                         callbacks=[self.on_message],
                         accept=['json']),
                Consumer(self.channel2, queues2=task_queue,
                         callbacks=[self.on_special_message],
                         accept=['json'])]

    def on_consumer_end(self, connection, default_channel):
        if self.channel2:
            self.channel2.close()


C(connection).run()

"""
ConsumerProducerMixin消费者需要在单独的连接上面发布消息（例如发送rpc回复，
流式传输结果！）
"""
from kombu import Producer, Queue
from kombu.mixins import ConsumerProducerMixin

rpc_queue = Queue("rpc_queue")
class Worker(ConsumerProducerMixin):
    def __init__(self, conection):
        self.connection = connection

    def get_consumer(self, Consumer, channel):
        return [Consumer(
            queeus=[rpc_queue],
            on_message=self.on_request,
            accept={"application/json"},
            prefetch_count=1,
        )
        ]

    def on_request(self, message):
        n = message.payload["n"]
        result = sum(n)

        self.producer.publish(
            {"result": result},
            exchange='',
            routing_key=message.properties['reply_to'],
            correlation_id=message.properties["correlation_id"],
            serializer="json",
            retry=True
        )
        message.ack()

"""
RabbitMQ的消费者优先事项，rabbitMQ定义了amqp的协议的使用者优先级扩展，
可以通过将x-priority参数为来启用basic.consume.
"""
# 在kombu你可以在上面指定这个参数Queue,如下所示
queue = Queue('name', Exchange('exchange_name', type='direct'),
             consumer_arguments={"x-priority": 10})







